package com.eshipenet.shipnet.edgeboxextbuilder
import org.apache.flink.api.common.state.ValueStateDescriptor
import org.apache.flink.streaming.api.functions.co.{CoMapFunction, CoProcessFunction}
import org.apache.flink.util.Collector

class EdgeBoxModelActionCoProcessFunction
    extends CoProcessFunction[(String, Set[EdgeBox]), EdgeBoxModelAction, EdgeBoxExtAction] {

  lazy val stateDesc = new ValueStateDescriptor[(String, Set[EdgeBox])]("state", classOf[(String, Set[EdgeBox])])

  override def processElement1(
      value: (String, Set[EdgeBox]),
      ctx: CoProcessFunction[(String, Set[EdgeBox]), EdgeBoxModelAction, EdgeBoxExtAction]#Context,
      out: Collector[EdgeBoxExtAction]): Unit = {
    val state = getRuntimeContext.getState(stateDesc)
    state.update(value)
  }

  override def processElement2(
      value: EdgeBoxModelAction,
      ctx: CoProcessFunction[(String, Set[EdgeBox]), EdgeBoxModelAction, EdgeBoxExtAction]#Context,
      out: Collector[EdgeBoxExtAction]): Unit = {
    val state      = getRuntimeContext.getState(stateDesc)
    val currentOpt = Option(state.value())
    currentOpt.foreach { current =>
      println("current", current)
      value match {
        case EdgeBoxModelInsert(id, after) =>
          if (current._1 == id) {
            current._2.foreach { edgeBox =>
              val ext = EdgeBoxExtModelUpdate(edgeBox.id, edgeBox, Some(after))
              out.collect(ext)
            }
          }
        case EdgeBoxModelUpdate(id, before, after) =>
          if (current._1 == id) {
            current._2.foreach { edgeBox =>
              val ext = EdgeBoxExtModelUpdate(edgeBox.id, edgeBox, Some(after))
              out.collect(ext)
            }
          }
        case EdgeBoxModelDelete(id, _) =>
          if (current._1 == id) {
            current._2.foreach { edgeBox =>
              val ext = EdgeBoxExtModelUpdate(edgeBox.id, edgeBox, None)
              out.collect(ext)
            }
          }
      }
    }

  }
}
